#include "net_bus.h"
#include <unistd.h>
#include <iostream>
#include <thread>
#include <chrono>
#include "google/protobuf/dynamic_message.h"
#include "../com/simple_logger.h"
#include "../work_thread.h"
#include "metadata.h"

using namespace std;
//
void NetBus::Emit()
{
    for (auto it = work_thread_list_.begin(); it != work_thread_list_.end(); it++)
    {
        (*it)->Notify("e");
    }
}

void NetBus::Regist(WorkThread *work)
{
    mtx_.lock();
    work_thread_list_.push_back(work);
    mtx_.unlock();
}

void NetBus::Add(const char *header, int header_size, const char *data, int data_size)
{
    mtx_.lock();
    Metadata *meta = new Metadata(header, header_size, data, data_size);
    meta->set_ttl(work_thread_list_.size());
    metadata_list_.push_back(meta);
    if(metadata_list_.size() == 1)
    {
        this->Emit();
    }
    mtx_.unlock();
}

Metadata *NetBus::ReadFromHead()
{
    return metadata_list_.front();
}

void NetBus::ReadFinish()
{
    mtx_.lock();
    Metadata *m = metadata_list_.front();
    m->Decrement();
    if (m->get_ttl() == 0)
    {
        delete m;
        metadata_list_.pop_front();
        if(metadata_list_.size() > 0)
        {
            this->Emit();
        }
    }
    
    mtx_.unlock();
}